import java.io.PrintWriter;
import java.net.Socket;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DatabaseConnection {
    private static final String URL = "jdbc:mysql://localhost:3306/ry-vue";
    private static final String USER = "root";
    private static final String PASSWORD = "DzjcMysql@123456.";
    private static final String SERVER_HOST = "192.168.0.200";
    private static final int SERVER_PORT = 9096;

    static class KafkaData {
        String epoch;
        String xecef;
        String yecef;
        String zecef;
        int monitorPointCode;

        KafkaData(String epoch, String xecef, String yecef, String zecef, int monitorPointCode) {
            this.epoch = epoch;
            this.xecef = xecef;
            this.yecef = yecef;
            this.zecef = zecef;
            this.monitorPointCode = monitorPointCode;
        }

        @Override
        public String toString() {
            return "epoch: " + (epoch != null ? epoch : "null") + 
                   ", xecef: " + (xecef != null ? xecef : "null") + 
                   ", yecef: " + (yecef != null ? yecef : "null") + 
                   ", zecef: " + (zecef != null ? zecef : "null") + 
                   ", monitorPointCode: " + monitorPointCode;
        }
    }

    public static void main(String[] args) {
        Connection connection = null;
        List<KafkaData> dataList = new ArrayList<>();

        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection(URL, USER, PASSWORD);
            System.out.println("数据库连接成功！");

            String selectSQL = "SELECT epoch, xecef, yecef, zecef, MONITORPOINTCODE FROM surf_kafkadata WHERE MONITORPOINTCODE = 12";
            PreparedStatement selectStmt = connection.prepareStatement(selectSQL);
            ResultSet resultSet = selectStmt.executeQuery();

            while (resultSet.next()) {
                String epoch = resultSet.getString("epoch");
                String xecef = resultSet.getString("xecef");
                String yecef = resultSet.getString("yecef");
                String zecef = resultSet.getString("zecef");
                int monitorPointCode = resultSet.getInt("MONITORPOINTCODE");
                dataList.add(new KafkaData(epoch, xecef, yecef, zecef, monitorPointCode));
            }
            System.out.println("读取到 " + dataList.size() + " 条 MONITORPOINTCODE = 12 的数据");

        } catch (ClassNotFoundException e) {
            System.out.println("未找到 JDBC 驱动: " + e.getMessage());
        } catch (SQLException e) {
            System.out.println("数据库操作失败: " + e.getMessage());
        } finally {
            try {
                if (connection != null && !connection.isClosed()) {
                    connection.close();
                    System.out.println("数据库连接已关闭");
                }
            } catch (SQLException e) {
                System.out.println("关闭连接失败: " + e.getMessage());
            }
        }

        if (!dataList.isEmpty()) {
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            final int[] index = {0};

            scheduler.scheduleAtFixedRate(() -> {
                if (index[0] < dataList.size()) {
                    KafkaData data = dataList.get(index[0]);
                    sendToServer(data.toString());
                    index[0]++;
                } else {
                    System.out.println("所有数据已发送完毕");
                    scheduler.shutdown();
                }
            }, 0, 3, TimeUnit.SECONDS);
        } else {
            System.out.println("没有数据可发送");
        }
    }

    private static void sendToServer(String data) {
        try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT);
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
            out.println(data);
            System.out.println("已发送到 " + SERVER_HOST + ":" + SERVER_PORT + ": " + data);
        } catch (Exception e) {
            System.out.println("发送失败: " + e.getMessage());
        }
    }
}